/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.bcps;

//import cn.ac.iie.di.bcps.JsonBcpsSender.ClientPool;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.session.REAbstractSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.RESessionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSessionBuilder;
import cn.ac.iie.di.global.GlobalParas;
import static cn.ac.iie.di.global.GlobalParas.connectionHandler;
//import static cn.ac.iie.di.global.GlobalParas.pcInfo;
import iie.mm.client.ClientAPI;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.NoSuchElementException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.logging.Level;
import java.util.logging.Logger;
import redis.clients.jedis.Jedis;

/**
 *
 * @author lenovo-pc
 */
public class JsonBcpsSender implements Runnable {

    private int addCount = 0;
    private long startTime = System.currentTimeMillis();
    static org.apache.log4j.Logger logger = null;

    static {
        logger = org.apache.log4j.Logger.getLogger(JsonBcpsSender.class.getName());
    }

    ThreadPoolExecutor threadpool;

    public JsonBcpsSender() {
        //创建一个线程池
        threadpool = new ThreadPoolExecutor(32, 32, 3, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(500),
                new ThreadFactory() {
            AtomicInteger idx = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "JsonBcpTask_" + idx.getAndIncrement());
            }
        }, new RejectedExecutionHandler() {

            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                while (true) {
                    try {
                        executor.getQueue().put(r);
                        break;
                    } catch (InterruptedException ex) {
                        logger.warn("", ex);
                    }
                }
            }
        });
    }

    @Override
    public void run() {

        logger.info("It is running!");
//        ClientPool pool;
//        try {
//            pool = new ClientPool();
//        } catch (Exception ex) {
//            logger.error("init session error :" + ex, ex);
//            return;
//        }
        try {
            String json = null;
            ArrayList<String> jsonList;
            while (true) {

                jsonList = new ArrayList<>(50);
                for (int i = 0; i < 50; ++i) {
                    //logger.info("try "+i);
                    String str = GlobalParas.newsQueue.poll(500, TimeUnit.MILLISECONDS);
                    if (str == null) {
                        break;
                    }
                    jsonList.add(str);
                }
                if (jsonList.isEmpty()) {
                    Thread.sleep(5000);
                    continue;
                }
                threadpool.execute(new JsonBcpsTask1(jsonList));
                ++addCount;

            }
        } catch (Exception ex) {
            logger.error("this user thread error :" + ex, ex);
        }

    }

    static public class ClientPool {

        static private final int poolSize = 3;

//        RESendSessionBuilder builder;
        LinkedBlockingQueue<ClientAPI> pool;
        int currentMax;
        Lock maxLock;

        public ClientPool() throws REConnectionException {
            try {
                //            this.builder = builder;
//                ClientAPI pcInfo = new ClientAPI();
//                pcInfo.init(GlobalParas.MEDIA_URL, GlobalParas.MEDIA_NAME);
                pool = new LinkedBlockingQueue<>();
                currentMax = 0;
                maxLock = new ReentrantLock();
            } catch (Exception ex) {
                logger.error(ex, ex);
            }
        }

        public synchronized ClientAPI borrowSession() throws RESessionException, REConnectionException {
            try {
                return pool.remove();
            } catch (NoSuchElementException ex) {
                if (currentMax < poolSize) {
                    maxLock.lock();
                    try {
                        if (currentMax < poolSize) {
                            ClientAPI pcInfo = new ClientAPI();
                            pcInfo.init(GlobalParas.MEDIA_URL, GlobalParas.MEDIA_NAME);
//                            Jedis edis = GlobalParas.connectionHandler.getJedis();
                            ++currentMax;
                            return pcInfo;
                        }
                    } catch (Exception ex1) {
                        logger.error(ex, ex);
                    } finally {
                        maxLock.unlock();
                    }
                }
                while (true) {
                    try {
                        return pool.take();
                    } catch (InterruptedException ex1) {
                    }
                }
            }
        }

        public synchronized void returnSession(ClientAPI session) {
            pool.add(session);
        }
    }

}
